package fnic.prehand.esagent;

import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.xcontent.json.JsonXContent;

import redis.clients.jedis.JedisPoolConfig;
import fnic.prehand.common.AgentConfig;


//public abstract class IndexAgent extends EsAgent implements Runnable{
public abstract class IndexAgent extends EsAgent{
	/* Redis Related things */
	protected static final AgentConfig config = AgentConfig.getInstance();
	private static final Logger logger = Logger.getLogger(IndexAgent.class);
	protected static volatile JsonXContent jsoncontent;
	private static final int esBulkSize = config.getclientBulkSize();
	
	protected boolean quickNeedIndex=false;
	protected boolean indexOver=false;
	protected String indexName;
	protected String typeName;
	private int sessionDoc = 0;
	
	private BulkRequestBuilder currentRequest;
//	private ThreadLocal<BulkRequestBuilder> currentRequest;
	
	static{
		jsoncontent = (JsonXContent) XContentFactory.xContent(XContentType.JSON);
	}
	
	public IndexAgent(String idx, String type){
		indexName = idx;
		typeName = type;
		currentRequest = client.prepareBulk();
	}
	
	
	public void atomProcess(Object jsonitem){
		String json = jsonAugment(jsonitem);
		if(json!=null){
			currentRequest.add(client.prepareIndex(indexName, typeName).setSource(json));
			quickNeedIndex = true;
			processBulkIfNeeded(false);
		}
	}
	

	
	public void processBulkIfNeeded(Boolean force) {
		if (!quickNeedIndex) {
			return;
		}
		int actionCount = currentRequest.numberOfActions();
		
		if (actionCount != 0 && (actionCount > esBulkSize || force == true)) {
			sessionDoc+=actionCount;
			try {
				BulkResponse response = currentRequest.execute().actionGet();
				dealBulkIndexRsp(response);
			} catch (Exception e) {
				logger.error("Failed to process bulk index", e);
			}
			quickNeedIndex = false;
			indexOver = force;
			currentRequest = client.prepareBulk();
			int agentSeqNum = getAgentSequenceNum();
			if(force == true){
				if(agentSeqNum!=-1){
					logger.info("passitive esgent["+agentSeqNum+"] bulk over, size=" + sessionDoc);
				}else{
					logger.info("es bulk over, size=" + sessionDoc);
				}
				sessionDoc = 0;
			}
		}
	}
	
	public int getAgentSequenceNum(){
		return -1;
	}
	
	protected abstract String jsonAugment(Object jsonitem);
	protected abstract void dealBulkIndexRsp(BulkResponse response);
//	protected abstract void dealTotalIndexRsp();

}
